package com.tcoiss.message.service.impl;



import com.tcoiss.message.domain.MqCommonRegistry;
import com.tcoiss.message.domain.MqCommonRegistryData;
import com.tcoiss.message.domain.MqCommonRegistryMessage;
import com.tcoiss.message.domain.ReturnT;
import com.tcoiss.message.mapper.MqCommonRegistryDataMapper;
import com.tcoiss.message.mapper.MqCommonRegistryMapper;
import com.tcoiss.message.mapper.MqCommonRegistryMessageMapper;
import com.tcoiss.message.service.ICommonRegistryService;
import com.tcoiss.message.util.JacksonUtil;
import com.tcoiss.message.util.PropUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.web.context.request.async.DeferredResult;

import javax.annotation.Resource;
import java.io.File;
import java.util.*;
import java.util.concurrent.*;

/**
 * xxl native regsitry, borrowed from "xxl-rpc"
 *
 * @author xuxueli 2018-11-26
 */
@Service
public class CommonRegistryServiceImpl implements ICommonRegistryService, InitializingBean, DisposableBean {
    private static Logger logger = LoggerFactory.getLogger(CommonRegistryServiceImpl.class);


    @Resource
    private MqCommonRegistryMapper commonRegistryDao;
    @Resource
    private MqCommonRegistryDataMapper commonRegistryDataDao;
    @Resource
    private MqCommonRegistryMessageMapper commonRegistryMessageDao;

    @Value("${xxl.mq.registry.data.filepath}")
    private String registryDataFilePath;
    @Value("${xxl.mq.registry.beattime}")
    private int registryBeatTime;
    @Value("${xxl.mq.registry.accessToken}")
    private String accessToken;

    @Override
    public ReturnT<String> registry(String accessToken, List<MqCommonRegistryData> commonRegistryDataList) {
        // valid
        if (this.accessToken!=null && this.accessToken.trim().length()>0 && !this.accessToken.equals(accessToken)) {
            return new ReturnT<String>(ReturnT.FAIL_CODE, "AccessToken Invalid");
        }
        if (commonRegistryDataList==null || commonRegistryDataList.size()==0) {
            return new ReturnT<String>(ReturnT.FAIL_CODE, "RegistryData Invalid.");
        }
        for (MqCommonRegistryData registryData: commonRegistryDataList) {
            if (registryData.getKey()==null || registryData.getKey().trim().length()==0 || registryData.getKey().trim().length()>255) {
                return new ReturnT<String>(ReturnT.FAIL_CODE, "RegistryData Key Invalid[0~255]");
            }
            if (registryData.getValue()==null || registryData.getValue().trim().length()==0 || registryData.getValue().trim().length()>255) {
                return new ReturnT<String>(ReturnT.FAIL_CODE, "RegistryData Value Invalid[0~255]");
            }
        }

        // add queue
        registryQueue.addAll(commonRegistryDataList);

        return ReturnT.SUCCESS;
    }

    @Override
    public ReturnT<String> remove(String accessToken, List<MqCommonRegistryData> commonRegistryDataList) {
        // valid
        if (this.accessToken!=null && this.accessToken.trim().length()>0 && !this.accessToken.equals(accessToken)) {
            return new ReturnT<String>(ReturnT.FAIL_CODE, "AccessToken Invalid");
        }
        if (commonRegistryDataList==null || commonRegistryDataList.size()==0) {
            return new ReturnT<String>(ReturnT.FAIL_CODE, "RegistryData Invalid.");
        }
        for (MqCommonRegistryData registryData: commonRegistryDataList) {
            if (registryData.getKey()==null || registryData.getKey().trim().length()==0 || registryData.getKey().trim().length()>255) {
                return new ReturnT<String>(ReturnT.FAIL_CODE, "RegistryData Key Invalid[0~255]");
            }
            if (registryData.getValue()==null || registryData.getValue().trim().length()==0 || registryData.getValue().trim().length()>255) {
                return new ReturnT<String>(ReturnT.FAIL_CODE, "RegistryData Value Invalid[0~255]");
            }
        }

        // add queue
        removeQueue.addAll(commonRegistryDataList);

        return ReturnT.SUCCESS;
    }

    @Override
    public ReturnT<Map<String, List<String>>> discovery(String accessToken, List<String> keys) {
        // valid
        if (this.accessToken!=null && this.accessToken.trim().length()>0 && !this.accessToken.equals(accessToken)) {
            return new ReturnT<>(ReturnT.FAIL_CODE, "AccessToken Invalid");
        }
        if (keys==null || keys.size()==0) {
            return new ReturnT<>(ReturnT.FAIL_CODE, "keys Invalid.");
        }
        for (String key: keys) {
            if (key==null || key.trim().length()==0 || key.trim().length()>255) {
                return new ReturnT<>(ReturnT.FAIL_CODE, "Key Invalid[0~255]");
            }
        }

        Map<String, List<String>> result = new HashMap<String, List<String>>();
        for (String key: keys) {
            MqCommonRegistryData commonRegistryData = new MqCommonRegistryData();
            commonRegistryData.setKey(key);

            List<String> dataList = new ArrayList<String>();
            MqCommonRegistry fileCommonRegistry = getFileRegistryData(commonRegistryData);
            if (fileCommonRegistry!=null) {
                dataList = fileCommonRegistry.getDataList();
            }

            result.put(key, dataList);
        }

        return new ReturnT<Map<String, List<String>>>(result);
    }

    @Override
    public DeferredResult<ReturnT<String>> monitor(String accessToken, List<String> keys) {
        // init
        DeferredResult deferredResult = new DeferredResult(registryBeatTime * 3 * 1000L, new ReturnT<>(ReturnT.FAIL_CODE, "Monitor timeout."));

        // valid
        if (this.accessToken!=null && this.accessToken.trim().length()>0 && !this.accessToken.equals(accessToken)) {
            deferredResult.setResult(new ReturnT<>(ReturnT.FAIL_CODE, "AccessToken Invalid"));
            return deferredResult;
        }
        if (keys==null || keys.size()==0) {
            deferredResult.setResult(new ReturnT<>(ReturnT.FAIL_CODE, "keys Invalid."));
            return deferredResult;
        }
        for (String key: keys) {
            if (key==null || key.trim().length()==0 || key.trim().length()>255) {
                deferredResult.setResult(new ReturnT<>(ReturnT.FAIL_CODE, "Key Invalid[0~255]"));
                return deferredResult;
            }
        }

        // monitor by client
        for (String key: keys) {
            String fileName = parseRegistryDataFileName(key);

            List<DeferredResult> deferredResultList = registryDeferredResultMap.get(fileName);
            if (deferredResultList == null) {
                deferredResultList = new ArrayList<>();
                registryDeferredResultMap.put(fileName, deferredResultList);
            }

            deferredResultList.add(deferredResult);
        }

        return deferredResult;
    }

    /**
     * update Registry And Message
     */
    private void checkRegistryDataAndSendMessage(MqCommonRegistryData commonRegistryData){
        // data json
        List<MqCommonRegistryData> commonRegistryDataList = commonRegistryDataDao.findData(commonRegistryData.getKey());
        List<String> valueList = new ArrayList<>();
        if (commonRegistryDataList!=null && commonRegistryDataList.size()>0) {
            for (MqCommonRegistryData dataItem: commonRegistryDataList) {
                valueList.add(dataItem.getValue());
            }
        }
        String dataJson = JacksonUtil.writeValueAsString(valueList);

        // update registry and message
        MqCommonRegistry commonRegistry = commonRegistryDao.load(commonRegistryData.getKey());
        boolean needMessage = false;
        if (commonRegistry == null) {
            commonRegistry = new MqCommonRegistry();
            commonRegistry.setKey(commonRegistryData.getKey());
            commonRegistry.setData(dataJson);
            commonRegistryDao.add(commonRegistry);
            needMessage = true;
        } else {
            if (!commonRegistry.getData().equals(dataJson)) {
                commonRegistry.setData(dataJson);
                commonRegistryDao.update(commonRegistry);
                needMessage = true;
            }
        }

        if (needMessage) {
            // sendRegistryDataUpdateMessage (registry update)
            sendRegistryDataUpdateMessage(commonRegistry);
        }

    }

    /**
     * send RegistryData Update Message
     */
    private void sendRegistryDataUpdateMessage(MqCommonRegistry rpcRegistry){
        String registryUpdateJson = JacksonUtil.writeValueAsString(rpcRegistry);

        MqCommonRegistryMessage registryMessage = new MqCommonRegistryMessage();
        registryMessage.setData(registryUpdateJson);
        commonRegistryMessageDao.add(registryMessage);
    }

    // ------------------------ broadcase + file data ------------------------

    private ExecutorService executorService = Executors.newCachedThreadPool();
    private volatile boolean executorStoped = false;
    private volatile List<Integer> readedMessageIds = Collections.synchronizedList(new ArrayList<Integer>());

    private volatile LinkedBlockingQueue<MqCommonRegistryData> registryQueue = new LinkedBlockingQueue<MqCommonRegistryData>();
    private volatile LinkedBlockingQueue<MqCommonRegistryData> removeQueue = new LinkedBlockingQueue<MqCommonRegistryData>();
    private Map<String, List<DeferredResult>> registryDeferredResultMap = new ConcurrentHashMap<>();

    public static MqCommonRegistryData staticRegistryData;

    @Override
    public void afterPropertiesSet() throws Exception {

        /**
         * registry registry data         (client-num/10 s)
         */
        executorService.execute(new Runnable() {
            @Override
            public void run() {
                while (!executorStoped) {
                    try {
                        MqCommonRegistryData commonRegistryData = registryQueue.take();
                        if (commonRegistryData !=null) {

                            // refresh or add
                            int ret = commonRegistryDataDao.refresh(commonRegistryData);
                            if (ret == 0) {
                                commonRegistryDataDao.add(commonRegistryData);
                            }

                            // valid file status
                            MqCommonRegistry fileCommonRegistry = getFileRegistryData(commonRegistryData);
                            if (fileCommonRegistry!=null && fileCommonRegistry.getDataList().contains(commonRegistryData.getValue())) {
                                continue;     // "Repeated limited."
                            }

                            // checkRegistryDataAndSendMessage
                            checkRegistryDataAndSendMessage(commonRegistryData);
                        }
                    } catch (Exception e) {
                        if (!executorStoped) {
                            logger.error(e.getMessage(), e);
                        }
                    }
                }
            }
        });

        /**
         * remove registry data         (client-num/start-interval s)
         */
        executorService.execute(new Runnable() {
            @Override
            public void run() {
                while (!executorStoped) {
                    try {
                        MqCommonRegistryData commonRegistryData = removeQueue.take();
                        if (commonRegistryData != null) {

                            // delete
                            commonRegistryDataDao.deleteDataValue(commonRegistryData.getKey(), commonRegistryData.getValue());

                            // valid file status
                            MqCommonRegistry fileCommonRegistry = getFileRegistryData(commonRegistryData);
                            if (fileCommonRegistry!=null && !fileCommonRegistry.getDataList().contains(commonRegistryData.getValue())) {
                                continue;   // "Repeated limited."
                            }

                            // checkRegistryDataAndSendMessage
                            checkRegistryDataAndSendMessage(commonRegistryData);
                        }
                    } catch (Exception e) {
                        if (!executorStoped) {
                            logger.error(e.getMessage(), e);
                        }
                    }
                }
            }
        });

        /**
         * broadcase new one registry-data-file     (1/1s)
         *
         * clean old message   (1/10s)
         */
        executorService.execute(new Runnable() {
            @Override
            public void run() {
                while (!executorStoped) {
                    try {
                        // new message, filter readed
                        List<MqCommonRegistryMessage> messageList = commonRegistryMessageDao.findMessage(readedMessageIds);
                        if (messageList!=null && messageList.size()>0) {
                            for (MqCommonRegistryMessage message: messageList) {
                                readedMessageIds.add(message.getId());

                                // from registry、add、update、deelete，ne need sync from db, only write
                                MqCommonRegistry commonRegistry = JacksonUtil.readValue(message.getData(), MqCommonRegistry.class);

                                // default, sync from db （aready sync before message, only write）

                                // sync file
                                setFileRegistryData(commonRegistry);
                            }
                        }

                        // clean old message;
                        if (System.currentTimeMillis() % registryBeatTime ==0) {
                            commonRegistryMessageDao.cleanMessage(10);
                            readedMessageIds.clear();
                        }
                    } catch (Exception e) {
                        if (!executorStoped) {
                            logger.error(e.getMessage(), e);
                        }
                    }
                    try {
                        TimeUnit.SECONDS.sleep(1);
                    } catch (Exception e) {
                        if (!executorStoped) {
                            logger.error(e.getMessage(), e);
                        }
                    }
                }
            }
        });

        /**
         *  clean old registry-data     (1/10s)
         *
         *  sync total registry-data db + file      (1+N/10s)
         *
         *  clean old registry-data file
         */
        executorService.execute(new Runnable() {
            @Override
            public void run() {
                while (!executorStoped) {
                    try {

                        // + static registry
                        if (staticRegistryData != null) {
                            registryQueue.add(staticRegistryData);
                        }

                        // clean old registry-data in db
                        commonRegistryDataDao.cleanData(registryBeatTime * 3);

                        // + clean old registry in db
                        commonRegistryDao.cleanDead();

                        // sync registry-data, db + file
                        int offset = 0;
                        int pagesize = 1000;
                        List<String> registryDataFileList = new ArrayList<>();

                        List<MqCommonRegistry> registryList = commonRegistryDao.pageList(offset, pagesize);
                        while (registryList!=null && registryList.size()>0) {

                            for (MqCommonRegistry registryItem: registryList) {

                                // default, sync from db
                                List<MqCommonRegistryData> commonRegistryDataList = commonRegistryDataDao.findData(registryItem.getKey());
                                List<String> valueList = new ArrayList<String>();
                                if (commonRegistryDataList!=null && commonRegistryDataList.size()>0) {
                                    for (MqCommonRegistryData dataItem: commonRegistryDataList) {
                                        valueList.add(dataItem.getValue());
                                    }
                                }
                                String dataJson = JacksonUtil.writeValueAsString(valueList);

                                // check update, sync db
                                if (!registryItem.getData().equals(dataJson)) {
                                    registryItem.setData(dataJson);
                                    commonRegistryDao.update(registryItem);
                                }

                                // sync file
                                String registryDataFile = setFileRegistryData(registryItem);

                                // collect registryDataFile
                                registryDataFileList.add(registryDataFile);
                            }


                            offset += 1000;
                            registryList = commonRegistryDao.pageList(offset, pagesize);
                        }

                        // clean old registry-data file
                        cleanFileRegistryData(registryDataFileList);

                    } catch (Exception e) {
                        if (!executorStoped) {
                            logger.error(e.getMessage(), e);
                        }
                    }
                    try {
                        TimeUnit.SECONDS.sleep(registryBeatTime);
                    } catch (Exception e) {
                        if (!executorStoped) {
                            logger.error(e.getMessage(), e);
                        }
                    }
                }
            }
        });


    }

    @Override
    public void destroy() throws Exception {
        executorStoped = true;
        executorService.shutdownNow();
    }


    // ------------------------ file opt ------------------------

    // get
    public MqCommonRegistry getFileRegistryData(MqCommonRegistryData commonRegistryData){

        // fileName
        String fileName = parseRegistryDataFileName(commonRegistryData.getKey());

        // read
        Properties prop = PropUtil.loadProp(fileName);
        if (prop!=null) {
            MqCommonRegistry fileCommonRegistry = new MqCommonRegistry();
            fileCommonRegistry.setData(prop.getProperty("data"));
            fileCommonRegistry.setDataList(JacksonUtil.readValue(fileCommonRegistry.getData(), List.class));
            return fileCommonRegistry;
        }
        return null;
    }
    private String parseRegistryDataFileName(String key){
        // fileName
        String fileName = registryDataFilePath
                .concat(File.separator).concat(key)
                .concat(".properties");
        return fileName;
    }

    // set
    public String setFileRegistryData(MqCommonRegistry commonRegistry){

        // fileName
        String fileName = parseRegistryDataFileName(commonRegistry.getKey());

        // valid repeat update
        Properties existProp = PropUtil.loadProp(fileName);
        if (existProp != null && existProp.getProperty("data").equals(commonRegistry.getData())
                ) {
            return new File(fileName).getPath();
        }

        // write
        Properties prop = new Properties();
        prop.setProperty("data", commonRegistry.getData());

        PropUtil.writeProp(prop, fileName);

        logger.info(">>>>>>>>>>> xxl-mq, setFileRegistryData: key={}, data={}", commonRegistry.getKey(), commonRegistry.getData());


        // brocast monitor client
        List<DeferredResult> deferredResultList = registryDeferredResultMap.get(fileName);
        if (deferredResultList != null) {
            registryDeferredResultMap.remove(fileName);
            for (DeferredResult deferredResult: deferredResultList) {
                deferredResult.setResult(new ReturnT<>(ReturnT.FAIL_CODE, "Monitor key update."));
            }
        }

        return new File(fileName).getPath();
    }
    // clean
    public void cleanFileRegistryData(List<String> registryDataFileList){
        filterChildPath(new File(registryDataFilePath), registryDataFileList);
    }

    public void filterChildPath(File parentPath, final List<String> registryDataFileList){
        if (!parentPath.exists() || parentPath.list()==null || parentPath.list().length==0) {
            return;
        }
        File[] childFileList = parentPath.listFiles();
        for (File childFile: childFileList) {
            if (childFile.isFile() && !registryDataFileList.contains(childFile.getPath())) {
                childFile.delete();

                logger.info(">>>>>>>>>>> xxl-mq, cleanFileRegistryData, RegistryData Path={}", childFile.getPath());
            }
            if (childFile.isDirectory()) {
                if (parentPath.listFiles()!=null && parentPath.listFiles().length>0) {
                    filterChildPath(childFile, registryDataFileList);
                } else {
                    childFile.delete();
                }

            }
        }

    }


}
